package com.example.rabbitmqbootconsumer.receiver.direct;

import com.example.rabbitmqbootconsumer.dto.MessageDTO;
import com.google.gson.Gson;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareBatchMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;

import static com.example.rabbitmqbootconsumer.constant.MessageConstant.DIRECT_QUEUE_NAME;
import static com.example.rabbitmqbootconsumer.constant.MessageConstant.FANOUT_QUEUE_NAME1;

/**
 * 手动确认监听类
 *
 * @author 伯路平
 * @since 2023/6/20
 **/
@Component
@Slf4j
public class AckDirectReceiver implements ChannelAwareMessageListener {

    private static final Gson GSON = new Gson();

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            byte[] body = message.getBody();

            MessageDTO messageDTO = GSON.fromJson(new String(body, Charset.defaultCharset()), MessageDTO.class);
            log.info("收到来自{}队列的消息，messageDTO:{}", message.getMessageProperties().getConsumerQueue(),
                    messageDTO);
            if (DIRECT_QUEUE_NAME.equals(message.getMessageProperties().getConsumerQueue())) {
                log.info("直连队列");
            } else if (FANOUT_QUEUE_NAME1.equals(message.getMessageProperties().getConsumerQueue())) {
                log.info("广播");
            }
            // 手动确认是否可以被批处理 第二个参数为true则可以一次性处理，
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            // 拒绝消费当前消息，如果第二个参数为true，重新放到队列中，再次消费，如果为false，告诉服务器直接丢弃
            channel.basicReject(deliveryTag, false);
            log.error("处理失败，丢弃该条消息");
            e.printStackTrace();
        }
    }
}
